package cn.juque.datapipeline.helper;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.lang.Assert;
import cn.juque.common.constants.MessageEnum;
import cn.juque.common.constants.NumberConstant;
import cn.juque.common.constants.YesOrNoEnum;
import cn.juque.common.exception.AppException;
import cn.juque.datapipeline.api.enums.*;
import cn.juque.datapipeline.bo.WriteParamBO;
import cn.juque.datapipeline.config.PropertiesConfig;
import cn.juque.datapipeline.datasource.AbstractDatasource;
import cn.juque.datapipeline.entity.*;
import cn.juque.datapipeline.mapper.*;
import cn.juque.datapipeline.utils.PageUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.sql.Connection;
import java.util.Comparator;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * @author juque
 * @version 1.0.0
 * <ul>
 *     <li>TaskInfoHelper</li>
 * </ul>
 * @date 2023-04-04 22:14:46
 **/
@Slf4j
@Component("taskInfoHelper")
public class TaskInfoHelper {

    @Resource
    private TaskInfoMapper taskInfoMapper;

    @Resource
    private DatabaseInfoMapper databaseInfoMapper;

    @Resource
    private DatasourceHelper datasourceHelper;

    @Resource
    private TaskScriptMapper taskScriptMapper;

    @Resource
    private TaskFieldMappingInfoMapper taskFieldMappingInfoMapper;

    @Resource
    private TaskGroupInfoMapper taskGroupInfoMapper;

    @Resource
    private TaskGroupRelationMapper taskGroupRelationMapper;

    @Resource
    private TaskGroupHelper taskGroupHelper;

    @Resource
    private TaskExecuteInfoMapper taskExecuteInfoMapper;

    @Resource
    private PropertiesConfig propertiesConfig;

    @Async("dataPipelineExecutor")
    public void startTask(String groupId) {
        Assert.notEmpty(groupId);
        TaskGroupInfo taskGroupInfo = this.taskGroupInfoMapper.selectById(groupId);
        // 执行前检查
        Boolean beforeCheck = this.checkBeforeStart(taskGroupInfo);
        if (!Boolean.TRUE.equals(beforeCheck)) {
            return;
        }
        // 更新状态为：待执行
        String batchNo = DateUtil.format(new Date(), DatePattern.PURE_DATETIME_MS_PATTERN);
        boolean success = this.taskGroupHelper.updateStatus(groupId, batchNo, taskGroupInfo.getExecuteStatus(), GroupExecuteStatusEnum.WAIT.getCode());
        if (!success) {
            return;
        }
        try {
            this.taskGroupHelper.updateStatus(groupId, GroupExecuteStatusEnum.ING.getCode());
            TaskGroupRelation relationQuery = new TaskGroupRelation();
            relationQuery.setTaskGroupId(groupId);
            List<TaskGroupRelation> relations = this.taskGroupRelationMapper.queryGroupRelationList(relationQuery);
            List<String> taskIdList = relations.stream().sorted(Comparator.comparing(TaskGroupRelation::getSortNo))
                    .map(TaskGroupRelation::getTaskId).collect(Collectors.toList());
            TaskInfo taskInfo;
            long totalCount = 0;
            for (String id : taskIdList) {
                taskInfo = this.taskInfoMapper.selectById(id);
                if (TaskTypeEnum.CHECK.getCode().equals(taskInfo.getTaskType())
                        && !Boolean.TRUE.equals(this.runCheckTask(groupId, batchNo, taskInfo))) {
                    break;
                }
                // 执行采数
                totalCount = totalCount + this.runTask(groupId, batchNo, taskInfo);
            }
            this.taskGroupHelper.updateStatus(groupId, GroupExecuteStatusEnum.END.getCode());
            log.info("任务组【{}】,共传输数据量:{}", taskGroupInfo.getGroupName(), totalCount);
        } catch (Exception e) {
            log.info("任务组【{}】,执行异常，error={}", taskGroupInfo.getGroupName(), e.getMessage(), e);
            this.taskGroupHelper.updateStatus(groupId, GroupExecuteStatusEnum.FAIL.getCode());
        }
    }

    /**
     * 执行采数任务
     *
     * @param groupId  任务组ID
     * @param batchNo  批次号
     * @param taskInfo 任务信息
     * @return 写入的数量
     */
    public Long runTask(String groupId, String batchNo, TaskInfo taskInfo) {
        // 禁用状态不执行
        if (TaskStatusEnum.DISABLED.getCode().equals(taskInfo.getTaskStatus())) {
            return 0L;
        }
        TaskExecuteInfo taskExecuteInfo = this.saveExecute(groupId, taskInfo.getId(), batchNo);
        try {
            log.info("任务【{}】开始执行", taskInfo.getTaskName());
            // 获取读连接
            DatabaseInfo databaseInfo = this.databaseInfoMapper.selectById(taskInfo.getSourceDatabase());
            AbstractDatasource datasource = this.datasourceHelper.render(databaseInfo);
            Connection inConnection = (Connection) datasource.createConnection(databaseInfo);
            // 获取读数脚本
            TaskScript taskScript = this.taskScriptMapper.selectById(taskInfo.getScriptId());
            String script = taskScript.getScript();
            // 获取写的驱动
            DatabaseInfo targetDatabase = this.databaseInfoMapper.selectById(taskInfo.getTargetDatabase());
            AbstractDatasource targetDatasource = this.datasourceHelper.render(targetDatabase);
            // 目标清数
            if (YesOrNoEnum.YES.getCode().equals(taskInfo.getClearByBefore())) {
                targetDatasource.truncate(targetDatabase, taskInfo.getTargetTable());
            }
            DbEnum sourceDbEnum = DbEnum.forEnum(databaseInfo.getDbType());
            int pageNo = 0;
            long readCount = 0L;
            long writeCount = 0L;
            String pageScript;
            WriteParamBO writeParamBO = this.buildWriteParamBO(taskInfo, targetDatabase);
            while (true) {
                pageNo = pageNo + 1;
                pageScript = PageUtil.buildPageInfo(script, sourceDbEnum, pageNo, taskInfo.getDataOffset());
                List<Map<String, Object>> pageData = datasource.inData(inConnection, pageScript);
                if (CollUtil.isEmpty(pageData)) {
                    break;
                }
                writeParamBO.setDataList(pageData);
                // 统计
                readCount = readCount + pageData.size();
                writeCount = writeCount + targetDatasource.outData(writeParamBO);
            }
            this.updateExecuteStatus(taskExecuteInfo, TaskExecuteStatusEnum.END.getCode());
            log.info("任务【{}】共分页:{},每页:{},共传输数据量:{}，成功传输数据量:{}", taskInfo.getTaskName(), pageNo - 1, taskInfo.getDataOffset(), readCount, writeCount);
            return writeCount;
        } catch (Exception e) {
            this.updateExecuteStatus(taskExecuteInfo, TaskExecuteStatusEnum.FAIL.getCode());
            log.error("任务【{}】执行异常", taskInfo.getTaskName(), e);
            throw new AppException(MessageEnum.SYSTEM_ERROR.getCode(), taskInfo.getTaskName() + "执行失败");
        }

    }

    /**
     * 执行预检任务
     *
     * @param groupId  任务组ID
     * @param batchNo  批次号
     * @param taskInfo 任务信息
     * @return Boolean
     */
    public Boolean runCheckTask(String groupId, String batchNo, TaskInfo taskInfo) {
        // 禁用状态默认预检成功
        if (TaskStatusEnum.DISABLED.getCode().equals(taskInfo.getTaskStatus())) {
            return true;
        }
        TaskExecuteInfo taskExecuteInfo = this.saveExecute(groupId, taskInfo.getId(), batchNo);
        try {
            // 获取读连接
            DatabaseInfo databaseInfo = this.databaseInfoMapper.selectById(taskInfo.getSourceDatabase());
            AbstractDatasource abstractDatasource = this.datasourceHelper.render(databaseInfo);
            Connection inConnection = (Connection) abstractDatasource.createConnection(databaseInfo);
            // 获取读数脚本
            TaskScript taskScript = this.taskScriptMapper.selectById(taskInfo.getScriptId());
            DbEnum sourceDbEnum = DbEnum.forEnum(databaseInfo.getDbType());
            String script = PageUtil.buildPageInfo(taskScript.getScript(), sourceDbEnum, 1, NumberConstant.ONE_THOUSAND);
            List<Map<String, Object>> pageData = abstractDatasource.inData(inConnection, script);
            boolean flag = CollUtil.isNotEmpty(pageData);
            if (flag) {
                log.info("预检任务【{}】，预检成功", taskInfo.getTaskName());
            } else {
                log.info("预检任务【{}】，预检失败", taskInfo.getTaskName());
            }
            this.updateExecuteStatus(taskExecuteInfo, TaskExecuteStatusEnum.END.getCode());
            return flag;
        } catch (Exception e) {
            this.updateExecuteStatus(taskExecuteInfo, TaskExecuteStatusEnum.FAIL.getCode());
            log.error("任务【{}】执行异常", taskInfo.getTaskName(), e);
            throw new AppException(MessageEnum.SYSTEM_ERROR.getCode(), taskInfo.getTaskName() + "执行失败");
        }

    }

    /**
     * 保存一次执行状态
     *
     * @param groupId 任务组ID
     * @param taskId  任务ID
     * @param batchNo 批次号
     * @return TaskExecuteInfo
     */
    public TaskExecuteInfo saveExecute(String groupId, String taskId, String batchNo) {
        TaskExecuteInfo taskExecuteInfo = new TaskExecuteInfo();
        taskExecuteInfo.setTaskId(taskId);
        taskExecuteInfo.setTaskGroupId(groupId);
        taskExecuteInfo.setExecuteBatchNo(batchNo);
        taskExecuteInfo.setStartTime(new Date());
        taskExecuteInfo.setExecuteStatus(TaskExecuteStatusEnum.ING.getCode());
        taskExecuteInfo.setExecuteIp(this.propertiesConfig.getBindIp());
        this.taskExecuteInfoMapper.insert(taskExecuteInfo);
        return taskExecuteInfo;
    }

    /**
     * 变更执行状态
     *
     * @param taskExecuteInfo TaskExecuteInfo
     * @param status          执行状态
     */
    public void updateExecuteStatus(TaskExecuteInfo taskExecuteInfo, Integer status) {
        if (TaskExecuteStatusEnum.END.getCode().equals(status)) {
            taskExecuteInfo.setEndTime(new Date());
        }
        taskExecuteInfo.setExecuteStatus(status);
        this.taskExecuteInfoMapper.updateById(taskExecuteInfo);
    }

    /**
     * 执行任务组前的检查
     *
     * @param taskGroupInfo 任务组信息
     * @return boolean
     */
    private Boolean checkBeforeStart(TaskGroupInfo taskGroupInfo) {
        // 禁用状态不执行
        if (GroupExecuteStatusEnum.DISABLED.getCode().equals(taskGroupInfo.getExecuteStatus())) {
            return false;
        }
        // 避免重复执行
        return !GroupExecuteStatusEnum.ING.getCode().equals(taskGroupInfo.getExecuteStatus());
    }

    /**
     * 构建变量
     *
     * @param taskInfo 任务
     * @param database 写库的驱动
     * @return WriteParamBO
     */
    private WriteParamBO buildWriteParamBO(TaskInfo taskInfo, DatabaseInfo database) {
        WriteParamBO paramBO = new WriteParamBO();
        paramBO.setName(taskInfo.getTaskName());
        paramBO.setTableName(taskInfo.getTargetTable());
        paramBO.setOpTypeEnum(DataOpTypeEnum.forEnum(taskInfo.getTargetOperateType()));
        AbstractDatasource datasource = this.datasourceHelper.render(database);
        // 获取写的驱动
        Object connection = datasource.createConnection(database);
        paramBO.setConnection(connection);
        TaskFieldMappingInfo fieldQuery = new TaskFieldMappingInfo();
        fieldQuery.setTaskId(taskInfo.getId());
        List<TaskFieldMappingInfo> fieldList = this.taskFieldMappingInfoMapper.queryFieldMappingInfoList(fieldQuery);
        paramBO.setFieldList(fieldList);
        return paramBO;
    }
}
